package bigdata.monitor.binlog.consumer;

import bigdata.monitor.binlog.entity.MonitorRule;
import bigdata.monitor.binlog.entity.MonitorResult;
import bigdata.monitor.binlog.service.RuleService;
import bigdata.monitor.binlog.utils.CanalToDebeziumConverter;
import bigdata.monitor.binlog.utils.JsonPathFieldExtractor;
import bigdata.monitor.binlog.utils.JsonUtils;
import bigdata.monitor.binlog.utils.UnifiedRuleEngine;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONWriter;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;

/**
 * Binlog 消息消费者
 * - 从 Kafka 读取 Flink CDC 的 Binlog JSON（格式包含 table/op/before/after）
 * - 根据缓存的监控规则，逐条校验当前消息
 * - 若某条规则校验失败（即数据违规），则生成违规结果
 * - 将违规结果序列化为 JSON，并发送到 Kafka 违规事件 Topic（如 monitor-violation-events）
 *
 * 架构说明：
 * - 不再使用“批量 SQL 表达式”方式（如 CombinedRuleSqlBuilder）
 * - 改为“逐条规则独立校验”模式，由 UnifiedRuleEngine 自动路由：
 *     • 基础规则（TYPE/ENUM/LENGTH/PATTERN） → 使用 json-schema-validator
 *     • 自定义规则（CUSTOM） → 使用 SQLite 内存表达式求值
 */
@Slf4j
@Component
public class BinlogConsumer {

    @Autowired
    private RuleService ruleService;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${monitor.initial-topics}")
    private String initialTopics;

    @Value("${monitor.violation-topic:monitor-violation-events}")
    private String violationTopic;

    /**
     * 监听初始配置的 Kafka Topic，消费 Binlog JSON 消息
     *
     * @param binlogJsonMessage Flink CDC 输出的完整 Binlog JSON 字符串
     *                          示例: {"table":"user","op":"u","before":{...},"after":{...}}
     * @param ack 手动 ACK 控制，确保消息处理完成后才提交 offset（at-least-once 语义）
     */
    @KafkaListener(topics = "#{'${monitor.initial-topics}'.split(',')}", groupId = "${monitor.consumer-group-id}")
    public void consume(String binlogJsonMessage, Acknowledgment ack) {
        try {
            // 1. 从 Binlog JSON 中提取表名（如 "user"）
            // canal to debezium
            String debeziumJson = CanalToDebeziumConverter.convert(binlogJsonMessage);
            JSONObject obj = JSON.parseObject(debeziumJson);
            JSONObject source = obj.getJSONObject("source");
            String databaseName = source.getString("db");
            String tableName = source.getString("table");
            if (tableName == null || tableName.isEmpty() || databaseName.isEmpty()) {
                log.warn("Skip message: table name not found in JSON: {}", debeziumJson);
                ack.acknowledge();
                return;
            }

            // 2. 获取该表对应的所有监控规则（已按优先级排序，基础规则在前）
            String key = databaseName + "." + tableName;
            List<MonitorRule> rules = ruleService.getSortedRulesByKey(key);
            if (rules.isEmpty()) {
                // 无规则则跳过，但仍需提交 offset 避免堆积
                ack.acknowledge();
                return;
            }

            // 3. 【核心逻辑】逐条校验规则
            //    对每条规则，调用 UnifiedRuleEngine 进行独立校验
            //    如果返回 false，表示该规则校验失败（即数据违规）
            String batchId = UUID.randomUUID().toString();
            for (MonitorRule rule : rules) {
                try {
                    boolean passed = UnifiedRuleEngine.evaluate(debeziumJson, rule);
                    if (passed) {
                        // 4. 规则校验成功 → 生成违规结果

                        // 提取主键值（用于标识哪条记录违规）
                        String primaryKey = JsonUtils.getPrimaryKey(debeziumJson, rule.getPrimaryField());

                        // 根据 output_fields 配置，提取 before/after 的变更字段值
                        String changedFields = JsonPathFieldExtractor.extractChangedFields(debeziumJson, rule.getOutputFields());

                        // 构建违规结果对象
                        MonitorResult result = new MonitorResult();
                        result.setId(IdWorker.getId());
                        result.setDatabaseName(databaseName);
                        result.setTableName(tableName);
                        result.setPrimaryKey(primaryKey);
                        result.setBatchId(batchId); // 用于去重或追踪
                        result.setRuleName(rule.getRuleName());
                        result.setRuleCode(rule.getRuleCode());
                        result.setChangedFields(changedFields); // JSON 字符串，如 {"name":{"before":"A","after":"B"}}
                        result.setCreateTime(LocalDateTime.now());
                        // 序列化为 JSON 并发送到违规事件 Topic
                        String violationJson = JSON.toJSONString(result, JSONWriter.Feature.WriteMapNullValue);
                        kafkaTemplate.send(violationTopic, violationJson);

                        log.info("Rule violated: {} ({}) for table={}, pk={}",
                                rule.getRuleName(), rule.getRuleCode(), tableName, primaryKey);
                    }
                } catch (Exception e) {
                    // 单条规则异常不应中断整个消息处理
                    log.error("Error evaluating rule: {} ({}), table: {}, JSON preview: {}",
                            rule.getRuleName(), rule.getRuleCode(), tableName,
                            debeziumJson != null ? debeziumJson.substring(0, Math.min(200, debeziumJson.length())) : "null", e);
                    // 继续下一条规则
                }
            }

        } catch (Exception e) {
            // 整体消息处理异常（如 JSON 解析失败）
            log.error("Fatal error processing Kafka message", e);
            // 注意：即使异常，仍提交 offset（避免消息阻塞），但需监控告警
        } finally {
            // 手动提交 offset，确保 at-least-once 语义
            // 即使部分规则失败或异常，也认为消息“已处理”
            ack.acknowledge();
        }
    }
}
